package som.simon.reactor.sinks;

import reactor.core.publisher.Sinks;
import som.simon.reactor.SimonUtil;

import java.util.concurrent.LinkedBlockingQueue;
import java.util.stream.IntStream;

public class SinksMain {

    public static void main(String[] args) {

        // Sinks ： 接收器

        // Sinks.UnicastSpec unicast = Sinks.many().unicast();   // 单播     ：只能绑定单个订阅者
        // Sinks.MulticastSpec multicast = Sinks.many().multicast(); // 多播
        // Sinks.MulticastReplaySpec replay = Sinks.many().replay();  // 重播

        Sinks.Many<Integer> sink = Sinks.many().unicast()
                .onBackpressureBuffer(new LinkedBlockingQueue<Integer>(5)); // 被压队列

        new Thread(() -> {
            IntStream.rangeClosed(1, 10).forEach(ix -> {
                sink.tryEmitNext(ix);
                SimonUtil.sleep();
            });
        }).start();

        sink.asFlux()
                .map(SimonUtil::format)
                .subscribe(v -> System.out.println("v = " + v));

        // TODO 单播: 只能绑定单个订阅者,否则会报错
        // Sinks.many().unicast() sinks only allow a single Subscriber
        // sink.asFlux().subscribe();

        SimonUtil.pause();
    }

}
